package com.bjtu.boatmq.broker.service.outer;

import cn.boat.router.MetaZookeeper;
import cn.boat.router.broker.BrokerZookeeper;
import cn.boat.router.consumer.ConsumerMetaData;
import cn.boat.router.consumer.GetConsumerMetaDataImpl;
import cn.boat.router.models.TopicAndPartition;
import cn.boat.router.producer.ProducerZookeeper;
import cn.boat.router.utils.ZkClientUtils;
import com.bjtu.boatmq.broker.config.BrokerConfig;
import org.I0Itec.zkclient.ZkClient;

import java.util.List;

/**
 * broker与router的交互类
 */
public class Broker2RouterService {
    private static int brokerId;
    /* zookeeper相关类对象 */
    private static ZkClientUtils.ZKConfig zkConfig;
    private static ZkClient zkClient;
    private static BrokerZookeeper bz;
    /* 用于测试创建Topic */
    private static ProducerZookeeper pz;

    private static List<TopicAndPartition> topicAndPartitionList = null;
    private static List<ConsumerMetaData> consumerMetaDataList = null;
    private static GetConsumerMetaDataImpl consumerMetaDataServer = null;

    static {
        //  创建一次zkClient
        zkConfig = new ZkClientUtils.ZKConfig();
        zkClient = new ZkClient(zkConfig.zkConnect, zkConfig.zkSessionTimeoutMs,
                zkConfig.zkConnectionTimeoutMs, new ZkClientUtils.StringSerializer());
        bz = new BrokerZookeeper(zkClient, new MetaZookeeper(zkClient, "/boatMQ"));
        pz = new ProducerZookeeper(zkClient, new MetaZookeeper(zkClient, "/boatMQ"));
        consumerMetaDataServer = new GetConsumerMetaDataImpl();
    }

    /**
     * 注册broker节点
     * @return
     */
    public boolean registerBroker(){
        try {
            // 获取brokerID
            brokerId = BrokerConfig.getBrokerId();
            // 获取注册等待时间
            int registerWaitTime = BrokerConfig.getRegisterBrokerTimeoutMs();
            bz.registerBroker(BrokerConfig.getBrokerId(),
                              BrokerConfig.getBrokerHost(),
                              BrokerConfig.getBrokerPort());
            // 处理等待，超时返回false
            while (bz.getBrokerById(brokerId) == null){
                if(registerWaitTime > 0){
                    registerWaitTime -= 500;
                }else{
                    bz.close();
                    return false;
                }
                Thread.sleep(500);
            }
            // 测试注册topic
            registerTopic("topic-0",1,1);
        }catch (Exception e){
            e.printStackTrace();
            BrokerLogger.write("10033","注册broker失败");
            return false;
        }
        return true;
    }

    /**
     * 模拟producer创建topic
     * @param topic
     * @param partitionNum
     * @param replia
     * @return
     */
    public static boolean registerTopic(String topic, int partitionNum, int replia){
        // 调用sky的代码，注册topic
        try {
            pz.registerTopic(topic, partitionNum, replia);
        } catch (Exception e) {
            BrokerLogger.write(e);
            return false;
        }
        return true;
    }

    /**
     * 获取broker的元数据（根据BrokerId）
     * @return
     */
    public static List<TopicAndPartition> getBrokerMetaData(){
        try {
            topicAndPartitionList = bz.getBrokerMapById(BrokerConfig.getBrokerId());
            // System.out.println(topicAndPartitionList.toString());
        } catch (Exception e) {
            e.printStackTrace();
        }
        return topicAndPartitionList;
    }

    /**
     * 获取消费者订阅信息
     */
    public static List<ConsumerMetaData> getConsuemrMetaData() {
        // 获取消费者订阅信息
        try {
            consumerMetaDataList = consumerMetaDataServer.getConsumerMetaData();
            for (ConsumerMetaData consumerMetaData : consumerMetaDataList) {
                System.out.println(consumerMetaData);
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
        return consumerMetaDataList;
    }

    /**
     * 关闭broker-router链接
     * @throws InterruptedException
     */
    public static void shutdown() throws InterruptedException {
        // 关闭zookeeper
        try {
            bz.close();
        } catch (InterruptedException e) {
            BrokerLogger.write(e);
        }
    }

}
